package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @author SHUJIA
 */

public class Demo4MySource {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //使用自定义source
        DataStreamSource<Integer> my = env.addSource(new MySource());

        my.print();

        env.execute();

    }

    /**
     * 自定义source,实现SourceFunction接口，在run方法中读取数据
     */
    static class MySource implements SourceFunction<Integer> {
        /**
         * run： 方法再任务启动的时候执行一次，
         * 在run方法中读取任何的数据
         *
         * @param ctx 用于将读取到的数据发送到下游
         */
        @Override
        public void run(SourceContext<Integer> ctx) throws Exception {

            while (true) {
                //发送数据到下游
                ctx.collect(100);

                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            //用于任务取消时回收资源
        }
    }
}
